基于websocket实现将公网请求映射到本地的请求转发(服务端篇)

#基于websocket实现将公网请求映射到本地的请求转发(服务端篇)| 来源: 网络整理| 查看: 265

公网ip可以被互联网访问，而本机的私有ip则不被允许，如何实现将公网ip的请求映射到本机的室友ip呢，我们基于websocket技术，将访问服务器的请求全部转发到本机上

websocket实现的话，是需要分为两个端，第一个是放在公网的服务端，第二个是放在本机的客户端，客户端则需要主动的去请求公网的服务端，实现连接成功后，服务端会把所有的请求推给客户端，客户端处理完成后，则将处理结果返回给服务端，再由服务端返回给请求端

服务端代码：

commons-io commons-io 2.7 org.springframework.boot spring-boot-starter-websocket 2.6.2

技术框架的使用的是springboot框架，这里只展示有用的代码

websocket服务扫描注解类

@Target(value = ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) @Component public @interface WebSocketComponent { String value() default ""; }

websocket注册配置类

import org.springframework.beans.BeansException; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.config.annotation.EnableWebSocket; import org.springframework.web.socket.config.annotation.WebSocketConfigurer; import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry; import org.springframework.web.socket.server.standard.ServletServerContainerFactoryBean; import java.util.Map; /** * @Classname WebSocketConfig * @Description TODO * @Date 2023/3/7 0:34 * @Created by XiongXiong * @Author: X.I.O */ @EnableWebSocket @Configuration public class WebSocketConfig implements ApplicationContextAware, WebSocketConfigurer { ApplicationContext applicationContext ; @Override public void registerWebSocketHandlers(WebSocketHandlerRegistry webSocketHandlerRegistry) { Map map = this.applicationContext.getBeansWithAnnotation(WebSocketComponent.class); for (String key : map.keySet()) { Object value = map.get(key); if(value instanceof WebSocketServer){ WebSocketServer webSocketServer = (WebSocketServer) value; webSocketHandlerRegistry.addHandler(webSocketServer,key) .setAllowedOrigins("*");; } } } private static final int MAX_MESSAGE_SIZE = 20 * 1024; private static final long MAX_IDLE = 60 * 60 * 1000; @Bean public ServletServerContainerFactoryBean createWebSocketContainer() { ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean(); // 设置50MB container.setMaxTextMessageBufferSize(10 * 1024 * 1024); container.setMaxBinaryMessageBufferSize(10 * 1024 * 1024); System.out.println("加载设置内存大小了"); return container; } @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext=applicationContext; } }

websocket通用服务类编写

import lombok.extern.slf4j.Slf4j; import org.springframework.web.socket.BinaryMessage; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.handler.AbstractWebSocketHandler; /** * @Classname WebSocketServer * @Description TODO * @Date 2023/3/7 0:46 * @Created by XiongXiong * @Author: X.I.O */ @Slf4j public class WebSocketServer extends AbstractWebSocketHandler { @Override public void afterConnectionEstablished(WebSocketSession session) throws Exception { log.info("建立连接"); } /** * 获取客户端来的消息 * @param session * @param message * @throws Exception */ @Override protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception { // 获得客户端传来的消息 log.info("接收消息"); } @Override protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception { log.info("发送二进制消息"); } @Override public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception { log.info("因为异常关闭连接"); } @Override public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { log.info("正常关闭连接"); } }

websocketmodel模型

import org.springframework.web.socket.BinaryMessage; import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; /** * @Classname WebSocketModel * @Description TODO * @Date 2023/3/8 11:04 * @Created by XiongXiong * @Author: X.I.O */ public class WebSocketModel { private String sessionId; private WebSocketSession session; private TextMessage textMessage; private BinaryMessage binaryMessage; private volatile Boolean isRead; private volatile Boolean isFile; public Boolean getRead() { return isRead; } public void setRead(Boolean read) { isRead = read; } public String getSessionId() { return sessionId; } public void setSessionId(String sessionId) { this.sessionId = sessionId; } public WebSocketSession getSession() { return session; } public void setSession(WebSocketSession session) { this.session = session; } public TextMessage getTextMessage() { return textMessage; } public WebSocketModel(String sessionId, WebSocketSession session, TextMessage textMessage, Boolean isRead) { this.sessionId = sessionId; this.session = session; this.textMessage = textMessage; this.isRead = isRead; } public WebSocketModel(String sessionId, WebSocketSession session, BinaryMessage binaryMessage, Boolean isRead) { this.sessionId = sessionId; this.session = session; this.binaryMessage = binaryMessage; this.isRead = isRead; } public void setTextMessage(TextMessage textMessage) { this.textMessage = textMessage; } public WebSocketModel() { } public BinaryMessage getBinaryMessage() { return binaryMessage; } public void setBinaryMessage(BinaryMessage binaryMessage) { this.binaryMessage = binaryMessage; } public Boolean getFile() { return isFile; } public void setFile(Boolean file) { isFile = file; } }

websocket实现请求转发并将执行的返回结果返回给请求端

import com.alibaba.fastjson.JSONObject; import com.example.util.Results; import com.example.websocket.WebSocketComponent; import com.example.websocket.WebSocketModel; import com.example.websocket.WebSocketServer; import com.example.websocket.WebSocketUtils; import org.springframework.web.socket.BinaryMessage; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; import javax.websocket.OnMessage; import java.nio.ByteBuffer; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.locks.ReentrantLock; /** * @Classname WebSocketServer * @Description TODO * @Date 2023/3/7 0:55 * @Created by XiongXiong * @Author: X.I.O */ @WebSocketComponent("/wsconnection") public class WebSocketConnectionServer extends WebSocketServer { private static Map concurrentHashMap=new ConcurrentHashMap(); @Override public void afterConnectionEstablished(WebSocketSession session) throws Exception { WebSocketUtils.sendMessage(session,"服务器连接成功"); super.afterConnectionEstablished(session); } @Override protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception { System.out.println("接收消息"); if(concurrentHashMap.containsKey(session.getId())){ WebSocketModel webSocketModel=concurrentHashMap.get(session.getId()); webSocketModel.setTextMessage(message); webSocketModel.setRead(true); webSocketModel.setFile(false); }else { WebSocketModel webSocketModel=new WebSocketModel(session.getId(),session,message,false); webSocketModel.setFile(false); concurrentHashMap.put(session.getId(),webSocketModel); } } @Override protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception { System.out.println("接收到文件数据了: "+message.getPayloadLength()); if(concurrentHashMap.containsKey(session.getId())){ WebSocketModel webSocketModel=concurrentHashMap.get(session.getId()); webSocketModel.setBinaryMessage(message); webSocketModel.setRead(true); webSocketModel.setFile(true); }else { WebSocketModel webSocketModel=new WebSocketModel(session.getId(),session,message,false); webSocketModel.setFile(true); concurrentHashMap.put(session.getId(),webSocketModel); } } @Override public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception { System.out.println(exception.getMessage()+" 异常"); concurrentHashMap.remove(session.getId()); super.handleTransportError(session, exception); } @Override public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { concurrentHashMap.remove(session.getId()); System.out.println(status.getReason()+"原因"); super.afterConnectionClosed(session, status); } final transient ReentrantLock lock = new ReentrantLock(); public Object sendMessage(Object message) { try { lock.lock(); if(concurrentHashMap.isEmpty()){ JSONObject obj=Results.error("暂无连接"); return obj.toJSONString(); } //默认取第一个 String sessionId=null; for (Map.Entry map : concurrentHashMap.entrySet()){ sessionId=map.getKey(); break; } WebSocketModel webSocketModel=concurrentHashMap.get(sessionId); WebSocketSession webSocketSession=webSocketModel.getSession(); //发送消息 WebSocketUtils.sendMessage(webSocketSession,message); String content=null; //默认十秒 Long endTimeMills=System.currentTimeMillis()+ 100000; webSocketModel=concurrentHashMap.get(sessionId); //如果是未读则等待 while (!webSocketModel.getRead()){ webSocketModel=concurrentHashMap.get(sessionId); if(webSocketModel.getRead()){ if(!webSocketModel.getFile()){ content=webSocketModel.getTextMessage().getPayload(); webSocketModel.setRead(false); return content; }else { ByteBuffer buffer=webSocketModel.getBinaryMessage().getPayload(); byte[] bytes=buffer.array(); webSocketModel.setRead(false); webSocketModel.setFile(false); return bytes; } } Long currentTimeMillis=System.currentTimeMillis(); if(currentTimeMillis-endTimeMills>0){ throw new Exception("返回连接超时"); } } //恰好可读了但循环中止 if(webSocketModel.getRead()){ if(!webSocketModel.getFile()){ content=webSocketModel.getTextMessage().getPayload(); webSocketModel.setRead(false); return content; }else { ByteBuffer buffer=webSocketModel.getBinaryMessage().getPayload(); webSocketModel.setRead(false); webSocketModel.setFile(false); byte[] bytes=buffer.array(); return bytes; } } return null; }catch (Exception e){ e.printStackTrace(); }finally { lock.unlock(); } return null; } }

websocket工具类

import com.alibaba.fastjson.JSONObject; import com.example.config.RequestModel; import org.springframework.web.socket.BinaryMessage; import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; import java.io.IOException; import java.nio.ByteBuffer; /** * @Classname WebSocketUtils * @Description TODO * @Date 2023/3/7 0:38 * @Created by XiongXiong * @Author: X.I.O */ public class WebSocketUtils { /** * 发送消息 * @param webSocketSession * @param msg */ public static void sendMessage(WebSocketSession webSocketSession, Object msg){ try { if(msg instanceof RequestModel){ String str= JSONObject.toJSONString(msg); System.out.println("发送消息了"+str); webSocketSession.sendMessage(new TextMessage(str)); } if(msg instanceof String){ webSocketSession.sendMessage(new TextMessage((String)msg)); } if(msg instanceof byte[]){ byte[] bytes=(byte[]) msg; BinaryMessage binaryMessage=new BinaryMessage(bytes); webSocketSession.sendMessage(binaryMessage); } if(msg instanceof ByteBuffer){ ByteBuffer byteBuffer=(ByteBuffer)msg; BinaryMessage binaryMessage=new BinaryMessage(byteBuffer); webSocketSession.sendMessage(binaryMessage); } }catch (IOException e){ e.printStackTrace(); } } }

如果springboot如果需要实现所有请求都转发的话，则需要先实现一个拦截器，代码示例如下

import org.springframework.context.annotation.Configuration; import org.springframework.web.servlet.config.annotation.CorsRegistry; import org.springframework.web.servlet.config.annotation.InterceptorRegistration; import org.springframework.web.servlet.config.annotation.InterceptorRegistry; import org.springframework.web.servlet.config.annotation.WebMvcConfigurer; /** * @Classname InterceptorConfiguration * @Description TODO * @Date 2023/3/8 17:43 * @Created by XiongXiong * @Author: X.I.O */ @Configuration public class InterceptorConfiguration implements WebMvcConfigurer { /** * 重写addCorsMappings()解决跨域问题 * 配置：允许http请求进行跨域访问 * * @param registry */ @Override public void addCorsMappings(CorsRegistry registry) { //指哪些接口URL需要增加跨域设置 registry.addMapping("/**") .allowedOrigins("*")//指的是前端哪些域名被允许跨域 //.allowedOriginPatterns("*") //需要带cookie等凭证时，设置为true，就会把cookie的相关信息带上 .allowCredentials(true) //指的是允许哪些方法 .allowedMethods("GET", "HEAD", "POST", "PUT", "DELETE", "OPTIONS") //cookie的失效时间，单位为秒（s），若设置为-1，则关闭浏览器就失效 .maxAge(3600); } /** * 重写addInterceptors()实现拦截器 * 配置：要拦截的路径以及不拦截的路径 * * @param registry */ @Override public void addInterceptors(InterceptorRegistry registry) { //注册Interceptor拦截器(Interceptor这个类是我们自己写的拦截器类) InterceptorRegistration registration = registry.addInterceptor(new Interceptor()); //addPathPatterns()方法添加需要拦截的路径 //所有路径都被拦截 registration.addPathPatterns("/**"); //excludePathPatterns()方法添加不拦截的路径 //添加不拦截路径 registration.excludePathPatterns("/wsconnection"); } }

拦截器里面具体的代码

import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.example.server.WebSocketConnectionServer; import com.example.util.Results; import com.example.util.SpringContextUtils; import lombok.extern.slf4j.Slf4j; import org.springframework.web.servlet.HandlerInterceptor; import javax.servlet.ServletOutputStream; import javax.servlet.ServletResponse; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; import javax.servlet.http.Part; import java.io.*; import java.util.Collection; import java.util.Enumeration; import java.util.HashMap; import java.util.Map; /** * @Classname Interceptor * @Description TODO * @Date 2023/3/8 17:45 * @Created by XiongXiong * @Author: X.I.O */ @Slf4j public class Interceptor implements HandlerInterceptor { /** * 在请求处理之前进行调用（Controller方法调用之前） */ @Override public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) { try { System.out.println(request.getRequestURI()+" "+request.getRemoteAddr()); System.out.println(request.getMethod()); // System.out.println(request.getParts()); Map heards=new HashMap(); Enumeration requestNames=request.getHeaderNames(); while (requestNames.hasMoreElements()) { String headName=requestNames.nextElement(); String headValue=request.getHeader(headName); heards.put(headName,headValue); } String queryString=request.getQueryString(); System.out.println(queryString); String getScheme=request.getScheme(); System.out.println(getScheme); // if(request.getParts()) // Collection parts=request.getParts(); // if(parts != null && parts.size() > 0){ // System.out.println(parts.size()); // for (Part part : parts) { // InputStream inputStream=part.getInputStream(); // System.out.println(part.getName()); // System.out.println(part.getSubmittedFileName()); // System.out.println(part.getContentType()); // } // } Map requestMap=new HashMap(); Map map=request.getParameterMap(); for (Map.Entry entry : map.entrySet()) { String key=entry.getKey(); String[] value=entry.getValue(); if(value != null && value.length > 0){ requestMap.put(key,value[0]); } } System.out.println(JSON.toJSONString(map)); StringBuffer jb = new StringBuffer(); String line = null; try { if (!(request instanceof RequestWrapper)) { request = new RequestWrapper(request); } BufferedReader reader = request.getReader(); while ((line = reader.readLine()) != null) { jb.append(line); } //获取websocket服务对象 WebSocketConnectionServer webSocketConnectionServer= SpringContextUtils.getBean(WebSocketConnectionServer.class); RequestModel requestModel=new RequestModel(); if(!"".equals(jb.toString())){ JSONObject json=JSONObject.parseObject(jb.toString()); requestModel.setBody(json); } requestModel.setCode(200); requestModel.setMethod(request.getMethod()); requestModel.setHearders(heards); requestModel.setScheme(getScheme); requestModel.setUrl(request.getRequestURI()); requestModel.setQueryString(queryString); requestModel.setRemoteAddress(request.getRemoteAddr()); requestModel.setRemoteHost(request.getRemoteHost()); requestModel.setQueryMap(requestMap); //获取到转发的请求数据 Object obj=webSocketConnectionServer.sendMessage(requestModel); if(obj instanceof String){ String objStr=(String) obj; responseJson(response,objStr); } if(obj instanceof byte[]){ byte[] bytes=(byte[]) obj; responseByte(response,bytes); } System.out.println(jb.toString()); }catch (IOException e) { e.printStackTrace(); } return false; } catch (Exception e) { e.printStackTrace(); } JSONObject obj=Results.error("请求超时"); responseJson(response,obj.toJSONString()); //如果设置为false时，被请求时，拦截器执行到此处将不会继续操作 return false; //如果设置为true时，请求将会继续执行后面的操作 } public static void responseJson(ServletResponse response, Map resultMap){ PrintWriter out = null; try { response.setCharacterEncoding("UTF-8"); response.setContentType("application/json"); out = response.getWriter(); out.println(JSON.toJSONString(resultMap)); } catch (Exception e) { System.out.println(e.getMessage()); }finally{ if(out!=null){ out.flush(); out.close(); } } } public static void responseByte(HttpServletResponse response,byte[] bytes){ ServletOutputStream out=null; InputStream is=null; try { response.setContentType("image/jpeg"); response.setHeader( "Content-disposition", "attachment; filename=test.jpeg"); out = response.getOutputStream(); is = new ByteArrayInputStream(bytes); byte[] buff = new byte[1024]; int len = 0; while((len=is.read(buff))!=-1){ out.write(buff, 0, len); } } catch (Exception e) { System.out.println(e.getMessage()); }finally{ if(out!=null) { try { out.flush(); out.close(); } catch (IOException e) { e.printStackTrace(); } } } } public static void responseJson(ServletResponse response, String str){ PrintWriter out = null; try { response.setCharacterEncoding("UTF-8"); response.setContentType("application/json"); out = response.getWriter(); out.println(str); } catch (Exception e) { System.out.println(e.getMessage()); }finally{ if(out!=null){ out.flush(); out.close(); } } } }

SpringContextUtils具体的代码

Results工具类

package com.example.util; import com.alibaba.fastjson.JSONObject; /** * @author X.I.O * @title: Results * @projectName springboot * @description: TODO * @date 2021/10/27 17:04 */ public class Results { private static JSONObject getJSONObject(){ JSONObject obj = new JSONObject(); String time=DateUtils.getTimeStr(); obj.put("time",time); return obj; } public static JSONObject success(){ JSONObject object= getJSONObject(); object.put(Contants.CODE,0); object.put(Contants.MSG,Contants.SUCCESS); return object; } public static JSONObject success(String msg){ JSONObject object=success(); object.put(Contants.MSG,msg); return object; } public static JSONObject success(String msg,Object data){ JSONObject object = success(); object.put(Contants.MSG,msg); object.put(Contants.DATA,data); return object; } public static JSONObject success(Object data){ JSONObject object=success(); object.put("data",data); return object; } public static JSONObject success(Object data,Object total){ JSONObject object=success(); object.put(Contants.DATA,data); object.put(Contants.TOTAL,total); return object; } public static JSONObject error(){ JSONObject object= getJSONObject(); object.put(Contants.CODE,1); object.put(Contants.MSG,Contants.ERROR); return object; } public static JSONObject error(Integer code,String msg){ JSONObject object=error(); object.put(Contants.CODE,code); object.put(Contants.MSG,msg); return object; } public static JSONObject error(String msg){ JSONObject object=error(); object.put(Contants.MSG,msg); return object; } }

实现对HttpServletRequest进行二次读取

import org.apache.commons.io.IOUtils; import javax.servlet.ReadListener; import javax.servlet.ServletInputStream; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletRequestWrapper; import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; /** * @Classname RequestWrapper * @Description TODO * @Date 2023/3/8 18:00 * @Created by XiongXiong * @Author: X.I.O */ public class RequestWrapper extends HttpServletRequestWrapper { private volatile byte[] body; public RequestWrapper(HttpServletRequest request) throws IOException { super(request); body = IOUtils.toByteArray(super.getInputStream()); } @Override public BufferedReader getReader() throws IOException { return new BufferedReader(new InputStreamReader(getInputStream())); } @Override public ServletInputStream getInputStream() throws IOException { return new RequestBodyCachingInputStream(body); } private class RequestBodyCachingInputStream extends ServletInputStream { private byte[] body; private int lastIndexRetrieved = -1; private ReadListener listener; public RequestBodyCachingInputStream(byte[] body) { this.body = body; } @Override public int read() throws IOException { if (isFinished()) { return -1; } int i = body[lastIndexRetrieved + 1]; lastIndexRetrieved++; if (isFinished() && listener != null) { try { listener.onAllDataRead(); } catch (IOException e) { listener.onError(e); throw e; } } return i; } @Override public boolean isFinished() { return lastIndexRetrieved == body.length - 1; } @Override public boolean isReady() { return isFinished(); } @Override public void setReadListener(ReadListener listener) { if (listener == null) { throw new IllegalArgumentException("listener cann not be null"); } if (this.listener != null) { throw new IllegalArgumentException("listener has been set"); } this.listener = listener; if (!isFinished()) { try { listener.onAllDataRead(); } catch (IOException e) { listener.onError(e); } } else { try { listener.onAllDataRead(); } catch (IOException e) { listener.onError(e); } } } @Override public int available() throws IOException { return body.length - lastIndexRetrieved - 1; } @Override public void close() throws IOException { lastIndexRetrieved = body.length - 1; body = null; } } }

RequestModel，websocket请求客户端的请求模型

import com.alibaba.fastjson.JSONObject; import java.io.Serializable; import java.util.Map; /** * @Classname RequestModel * @Description TODO * @Date 2023/3/8 20:06 * @Created by XiongXiong * @Author: X.I.O */ public class RequestModel implements Serializable { private Integer code; private String url; private Map hearders; private String method; private Map queryMap; private JSONObject body; private String queryString; private String scheme; private String remoteAddress; private String remoteHost; public Integer getCode() { return code; } public void setCode(Integer code) { this.code = code; } public String getRemoteHost() { return remoteHost; } public void setRemoteHost(String remoteHost) { this.remoteHost = remoteHost; } public String getQueryString() { return queryString; } public void setQueryString(String queryString) { this.queryString = queryString; } public String getScheme() { return scheme; } public void setScheme(String scheme) { this.scheme = scheme; } public String getRemoteAddress() { return remoteAddress; } public void setRemoteAddress(String remoteAddress) { this.remoteAddress = remoteAddress; } public String getUrl() { return url; } public void setUrl(String url) { this.url = url; } public Map getHearders() { return hearders; } public void setHearders(Map hearders) { this.hearders = hearders; } public String getMethod() { return method; } public void setMethod(String method) { this.method = method; } public Map getQueryMap() { return queryMap; } public void setQueryMap(Map queryMap) { this.queryMap = queryMap; } public JSONObject getBody() { return body; } public void setBody(JSONObject body) { this.body = body; } public RequestModel(String url, Map hearders, String method, Map queryMap, JSONObject body) { this.url = url; this.hearders = hearders; this.method = method; this.queryMap = queryMap; this.body = body; } public RequestModel() { } }

服务端访问: ws://ip:port/wsconnection

客户端请看我写的另一篇

【本文地址】

公司简介

联系我们